package com.fanwei.rocketmq.service.impl;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.fanwei.rocketmq.common.RocketmqConstant;
import com.fanwei.rocketmq.domain.Test;
import com.fanwei.rocketmq.service.UserService;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;

/**
 * User消息执行Impl
 * @author fanwei
 */
@Service
public class UserServiceImpl implements UserService {

    @Resource(name = "rocketMQTemplate")
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 普通消费
     * @return
     */
    @Override
    public String test() {
        Test test = new Test();
        test.setUuid(IdUtil.fastUUID());
        test.setName("张三");
        test.setPhone("18888888888");
        rocketMQTemplate.convertAndSend(RocketmqConstant.TOPIC_TEST,test);
        return JSON.toJSONString(test,true);
    }

    /**
     * 顺序消费-通过haskey来区分发送到哪个queue中
     * @return
     */
    @Override
    public String orderHasKey() {
        for (int i=1; i<=5; i++){
            rocketMQTemplate.syncSendOrderly(RocketmqConstant.TOPIC_ORDER,"顺序消费orderOne:"+i, "orderOne");
        }
        for(int i=1; i<=5; i++){
            rocketMQTemplate.syncSendOrderly(RocketmqConstant.TOPIC_ORDER, "顺序消费orderTwo:"+i, "orderTwo");
        }
        return JSON.toJSONString("ok",true);
    }

    /**
     * 顺序消费-通过取出queue数量根据你的orderid进行取余来发送到哪个queue中
     * @return
     */
    @Override
    public String orderQueueNum() {
        rocketMQTemplate.setMessageQueueSelector((List<MessageQueue> mqs, Message msg, Object arg) -> {
            int queueNum = Integer.valueOf(String.valueOf(arg)) % mqs.size();
            System.out.println("队列id：" + queueNum + " 消息:" + new String(msg.getBody()));
            return mqs.get(queueNum);
        });
        for(int i=1; i<=10; i++){
            String msg = "type:" + i%4 + " value:" + i;
            rocketMQTemplate.syncSendOrderly(RocketmqConstant.TOPIC_ORDER, msg, String.valueOf(i));
        }
        return JSON.toJSONString("ok", true);
    }
}
